package com.econ.powercloud.ems.observer;

import com.econ.powercloud.clickhouse.entity.EmsFaultData;
import com.econ.powercloud.common.base.facade.util.TimeUtil;
import com.econ.powercloud.ems.common.enums.*;
import com.econ.powercloud.ems.common.interf.StatusAlarmFaultDataObserver;
import com.econ.powercloud.ems.common.pojo.EmsKafkaYxDataDTO;
import com.econ.powercloud.ems.common.util.IdWorker;
import com.econ.powercloud.ems.config.EMSStaticMap;
import com.econ.powercloud.ems.constant.SysConstant;
import com.econ.powercloud.ems.dal.entity.EmsDevice;
import com.econ.powercloud.ems.dal.entity.EmsFault;
import com.econ.powercloud.ems.pojo.EmsFaultRedisVO;
import com.econ.powercloud.ems.service.biz.ClickHouseBizService;
import com.econ.powercloud.ems.service.biz.FaultService;
import com.econ.powercloud.ems.util.RedisCacheUtil;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import org.apache.catalina.startup.RealmRuleSet;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Author : zjf
 * @Description : 判断遥信点号是否是故障告警信息
 * @Date : 2022/10/31 16:40
 * @Modified by : zjf
 */
@Component
public class StatusAlarmFaultDataProcessor implements StatusAlarmFaultDataObserver {

    @Autowired
    private IdWorker idWorker;
    @Autowired
    private FaultService faultService;
    @Autowired
    private ClickHouseBizService clickHouseBizService;
    @Autowired
    private RedisCacheUtil redisCacheUtil;
    @Resource(name="yxFaultPoolTaskExecutor")
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;

    @Override
    public void doInvoke(List<EmsKafkaYxDataDTO> dataList){
        //批量插入clickhouse的数据
        List<EmsFaultData> emsFaultList = Lists.newArrayList();
        //这里先做一次过滤,判断该点号是不是故障告警点号,不是不用处理
        List<EmsKafkaYxDataDTO> faultAlarmList = Lists.newArrayList();
        //pid和故障告警等级的map
        Map<String,Integer> pidToAlarmFaultCodeMap = Maps.newHashMap();
        dataList.stream().forEach(data ->{
            //先获取该设备对应的根设备是什么(这里的根设备指的是这个设备的moduleCode是def的那个设备,如PCU的根设备就是PCS)
            EmsDevice emsDevice = EMSStaticMap.moduleCodeToDeviceMap.get(data.getDeviceCode() + SysConstant.SPOT + (data.getModuleCode()!=null?data.getModuleCode():"def"));
            if(emsDevice == null){
                return;
            }
            //判断该点号是不是故障告警点号,不是不用处理
            String dit = data.getPid().replace(data.getDeviceCode() + SysConstant.SPOT, "");
            Map<String,Integer> faultDitMap = EMSStaticMap.faultDitMap.get(emsDevice.getProductKey());
            if(MapUtils.isEmpty(faultDitMap)){
                return;
            }
            Integer alarmFaultCode = faultDitMap.get(dit);
            if(alarmFaultCode == null){
                return;
            }
            pidToAlarmFaultCodeMap.put(data.getPid(),alarmFaultCode);
            faultAlarmList.add(data);
        });
        if(CollectionUtils.isEmpty(faultAlarmList)){
            return;
        }
        //更新redis中的故障告警最新一条缓存数据
        List<EmsFaultRedisVO> emsFaultRedisVOList = Lists.newArrayList();
        //获取遥信点号对应事件的最新一条记录,key:pid
        Map<String,EmsFault> leastFaultMap = faultService.selectLeastFaultList(faultAlarmList);
        for (EmsKafkaYxDataDTO data : faultAlarmList){
            //获取设备信息
            EmsDevice emsDevice = EMSStaticMap.moduleCodeToDeviceMap.get(data.getDeviceCode() + SysConstant.SPOT + (data.getModuleCode()!=null?data.getModuleCode():"def"));
            Integer alarmFaultCode = pidToAlarmFaultCodeMap.get(data.getPid());
            Integer val = Integer.parseInt(data.getVal());
            //查询该点号的最新一条数据
            EmsFault leastFaultRecord = leastFaultMap.get(data.getPid());
            /**
             * 1.若没有数据,记录当前数据
             * 2.如果上一次的值 与 这一次的值不同,就直接插入
             */
            if(leastFaultRecord == null || !leastFaultRecord.getFaultVal().equals(val)){
                EmsFaultData emsFault = this.assembleEmsFault(data,emsDevice,alarmFaultCode);
                emsFaultList.add(emsFault);
                emsFaultRedisVOList.add(this.assembleEmsFaultRedisVO(emsFault));
                continue;
            }
            //如果上一次的值是1,并且上一次的处理状态为已处理，就需要新插入数据
            if(YxValueType.break_position.getCode().equals(val) && EventDealStatusType.DEAL.getCode().equals(leastFaultRecord.getDealState())){
                EmsFaultData emsFault = this.assembleEmsFault(data,emsDevice,alarmFaultCode);
                emsFaultList.add(emsFault);
                emsFaultRedisVOList.add(this.assembleEmsFaultRedisVO(emsFault));
                continue;
            }
        }
        if(!CollectionUtils.isEmpty(emsFaultRedisVOList)){
            emsFaultRedisVOList.stream().collect(Collectors.groupingBy(d->d.getDeviceCode())).forEach((k,v)->{
                redisCacheUtil.addEmsFaultList(k, v);
            });
        }
        if(!CollectionUtils.isEmpty(emsFaultList)){
            //使用单线程池异步的处理插入clickhouse数据
            threadPoolTaskExecutor.execute(()-> clickHouseBizService.insertEmsFaultListData2ClickHouse(emsFaultList));
        }
    }

    private EmsFaultData assembleEmsFault(EmsKafkaYxDataDTO data, EmsDevice emsDevice, Integer alarmFaultCode){
        Integer val = Integer.parseInt(data.getVal());
        Integer faultType = AlarmOrFaultState.NORMAL.getCode();
        if(YxValueType.break_position.getCode().equals(val)){
            faultType = alarmFaultCode;
        }
        EmsFaultData emsFault = EmsFaultData.builder()
                .faultPid(data.getPid()).faultName(data.getRemark())
                .faultType(faultType)
                .faultResource(EventSourceType.YX.getCode()).energyStationId(emsDevice.getEnergyStationId())
                .energyEquipmentId(emsDevice.getEnergyEquipmentId()).deviceCode(emsDevice.getDeviceCode())
                .faultVal(val).dealState(YxValueType.break_position.getCode().equals(val)? EventDealStatusType.UN_DEAL.getCode():null)
                .realPid(data.getPid()).realPidVal(data.getVal()).realMeasureType(MeasureType.yx.getCode())
                .gmtCreate(data.getTime())
                .build();
        emsFault.setId(TimeUtil.format2String(emsFault.getGmtCreate(), DateTimeFormatter.ofPattern("yyyy"))+idWorker.nextId());
        return emsFault;
    }

    private EmsFaultRedisVO assembleEmsFaultRedisVO(EmsFaultData emsFault){
        return EmsFaultRedisVO.builder().faultPid(emsFault.getFaultPid()).faultVal(emsFault.getFaultVal())
                .deviceCode(emsFault.getDeviceCode()).faultResource(emsFault.getFaultResource())
                .dealState(emsFault.getDealState()).gmtCreate(emsFault.getGmtCreate()).build();
    }
}
